package com.zf.nospring.rabbitmq.delay.plugins;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.zf.nospring.rabbitmq.utils.RabbitUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

/**
 * 使用 https://github.com/rabbitmq/rabbitmq-delayed-message-exchange 插件 实现延迟消息，该插件提供了一个延迟交换机类型："x-delayed-message"。
 * <p>
 * 其大概原理就是：指定了延时的消息，会被先保存在 Mnesia （erlang编写的数据库管理系统）中，然后有一个定时器去查询最近需要被投递的消息，然后使用延迟交换机将其投递到目标队列中。
 * <p>
 * 但是该插件也有问题：
 * - 延时不可靠，存在消息数量较大或使用很久后延迟不准确（会推迟）
 * - 无备份机制，延时消息存在单个节点磁盘中
 * - 增加大量内存的占用 （经测试发现，发送大量延时消息后，rabbitmq内存占用明显增高，比普通消息还要高很多那种。）
 * 所以如果业务比较重要，还是不建议使用
 */
public class DelayPluginProducer {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitUtils.getDefaultConnection();
        Channel channel = connection.createChannel();

        // 声明一个延迟交换机
        HashMap<String, Object> exchangeArgs = new HashMap<>();
        // 指定用什么类型的交换机实现延迟交换机，这里使用direct类型的交换机表示在向延迟队列投递消息时按照direct的规则去匹配；也可以使用topic,fanout，headers
        exchangeArgs.put("x-delayed-type", "direct");
        // "x-delayed-message" 类型 表示延迟交换机
        String testPluginDelayExchange = "test_plugin_delay_exchange";
        channel.exchangeDeclare(testPluginDelayExchange, "x-delayed-message", true, false, exchangeArgs);

        // 声明一个普通队列，用作延迟队列
        String testPluginDelayQueue = "test_plugin_delay_queue";
        channel.queueDeclare(testPluginDelayQueue, true, false, false, null);

        // 绑定 延迟交换机和延迟队列
        String testPluginDelayRoutingKey = "test_plugin_delay_routing_key";
        channel.queueBind(testPluginDelayQueue, testPluginDelayExchange, testPluginDelayRoutingKey);


        HashMap<String, Object> queueHeaders = new HashMap<>();
        // 添加 "x-delay" 头，表示消息的延迟时间，单位是 毫秒，表示发出的消息经过多少毫秒后才会到达队列
        queueHeaders.put("x-delay", 10000);
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .headers(queueHeaders)
                .build();
        for (int i = 0; i < 5; i++) {
            // 向 延迟交换机发送消息
            channel.basicPublish(testPluginDelayExchange, testPluginDelayRoutingKey, properties, ("你好" + i).getBytes(StandardCharsets.UTF_8));
        }
        channel.close();
        connection.close();
    }
}
